package main;

import helper.CompositeKey;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
 
public class PairPartitioner extends Partitioner<CompositeKey, IntWritable> {
 
	HashPartitioner<Text, IntWritable> hashPartitioner = new HashPartitioner<Text, IntWritable>();
	Text newKey = new Text();
 
	@Override
	public int getPartition(CompositeKey key, IntWritable value, int numReduceTasks) {
	
		try {
			// Execute the default partitioner over the first part of the key
			newKey.set(String.valueOf(key.getUser()));
			return hashPartitioner.getPartition(newKey, value, numReduceTasks);
		} catch (Exception e) {
			e.printStackTrace();
			return Math.abs((int) (Math.random() * numReduceTasks)); // this would return a random value in the range
			// [0,numReduceTasks)
		}
	}
}
